package main

import (
	_ "GraduationProject/public/tools"
	"encoding/json"
	"flag"
	"os"
	"time"

	"GraduationProject/engine/producer"
	_ "GraduationProject/engine/producer"
	"GraduationProject/public/db"
	"GraduationProject/public/elastic_search"
	"GraduationProject/public/models"
	"GraduationProject/public/params"
	"GraduationProject/public/rpc_func"
	"GraduationProject/public/status"
	"GraduationProject/public/tools"
	"GraduationProject/rpcsupport"
	"fmt"
	"github.com/jinzhu/gorm"
	"net/rpc"
	"strconv"
)

var (
	storeClient *rpc.Client
	syncDB      = flag.Bool("syncdb", false, "sync database.")
)

func init() {
	flag.Parse()

	if *syncDB {
		db.SyncDB()
		os.Exit(0)
	}

	var err error
	db.ConnectDB()
	db.RedisConnect()
	db.ElasticSearchConnect()
	storeClient, err = rpcsupport.NewRpcClient(tools.Conf.Get("storage_host"))
	if err != nil {
		panic(err)
	}
	go loadSeeds()
}

/*
	爬虫引擎
*/
func main() {
	err := rpcsupport.ServeRpc(tools.Conf.Get("port"), &Engine{})
	if err != nil {
		panic(err)
	}
}

func loadSeeds() {
	var sleep = time.Second * 30
	for {
		var seeds []models.Seed
		var count int
		tx := db.DB.Begin()
		tx.Set("gorm:query_option", "FOR UPDATE").
			Where("`loaded` = ?", false).Find(&seeds)

		for _, v := range seeds {
			err := producer.Product(models.Record{
				SeedID:       v.ID,
				Url:          v.Url,
				Params:       v.Prams,
				FunctionName: v.FunctionName,
			})
			if err != nil {
				fmt.Println("loadSeeds error: ", err)
				continue
			}
			tx.Model(&models.Seed{}).
				Where("id = ?", v.ID).Update("loaded", true)
			count++
		}

		// 如果无修改，可能导致commit报错
		if count == 0 {
			tx.Rollback()
			time.Sleep(sleep)
			continue
		}
		if err := tx.Commit().Error; err != nil {
			tx.Rollback()
			fmt.Println("loadSeeds commit error: ", err)
			time.Sleep(sleep)
			continue
		}

		time.Sleep(sleep)
	}
}

type Engine struct{}

func (e *Engine) PushResult(result params.Result, message *params.Message) error {
	//fmt.Printf("Engine PushResult result:%#v\n", result)
	for _, s := range result.Seeds {
		if err := producer.Product(s); err != nil {
			_ = elastic_search.Elastic(db.ElasticClient, elastic_search.TypeEngine, elastic_search.DefaultLog{
				Level: elastic_search.LevelDefault,
				Log:   err.Error(),
			})
			continue
		}
	}
	var m params.Message
	bytes, _ := json.Marshal(result)
	fmt.Printf("storage.Call result: %s\n", bytes)
	if err := storeClient.Call(rpc_func.StorageSave, result, &m); err != nil {
		*message = m
		return nil
	}
	*message = params.Message{
		Status:  200,
		Message: "success",
	}
	return nil
}

func (e *Engine) NewWorkerID(worker models.Worker, message *params.Message) error {
	if worker.MachineID == 0 {
		*message = params.Message{
			Status:  status.MachineIDNotFound,
			Message: "无MachineID",
		}
		return nil
	}
	err := db.DB.Where("id = ?", worker.ID).First(&worker).Error
	if err == nil {
		*message = params.Message{
			Status:  status.WorkerIDNormal,
			Message: strconv.Itoa(int(worker.ID)),
		}
		return nil
	}

	db.Redis.HSet(models.WorkerStatusKey, strconv.Itoa(int(worker.ID)), models.WorkerPause)

	if err == gorm.ErrRecordNotFound {
		w := models.Worker{
			MachineID: worker.MachineID,
			Status:    models.WorkerRunning,
		}

		if err := db.DB.Create(&w).Error; err != nil {
			*message = params.Message{
				Status:  status.CreateWorkerIDError,
				Message: "生成WorkerID失败",
			}
			return nil
		}
		*message = params.Message{
			Status:  status.WorkerIDRevised,
			Message: strconv.Itoa(int(w.ID)),
		}
		return nil
	}

	*message = params.Message{
		Status:  status.ServerError,
		Message: "服务器异常",
	}
	return nil
}
